package cn.jly.bigdata.spark.core

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author lanyangji
 * @date 2019/11/28 15:35
 */
object Spark12_Mysql {

  val driver = "com.mysql.jdbc.Driver"
  val url = "jdbc:mysql:///rdd"
  val username = "root"
  val password = "123456"

  def main(args: Array[String]): Unit = {

    val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("test-file"))

    // 从mysql中读
    //    val jdbcRdd: JdbcRDD[(Int, String, Int)] = new JdbcRDD(
    //      sc,
    //      // 获取连接
    //      () => {
    //        Class.forName(driver)
    //        DriverManager.getConnection(url, username, password)
    //      },
    //      // sql语句
    //      "select * from user where id between ? and ?;",
    //      // 第一个参数
    //      1,
    //      // 第二个参数
    //      10,
    //      // 分区数
    //      1,
    //      // 结果集操作
    //      (r: ResultSet) => (r.getInt(1), r.getString(2), r.getInt(3))
    //    )
    //
    //    println(jdbcRdd.count())
    //    jdbcRdd.foreach(println)

    // 向mysql中写
    val listRdd: RDD[(String, Int)] = sc.makeRDD(List(("lanyangji", 26), ("wanghaitao", 26), ("wuting", 26)))

    // foreach是在executor中执行的，会发给不同的executor，直接带来的结果是，数据库中插入记录的顺序与定义的listRdd中不一致
    // 之所以不用foreach，是因为，如果用foreach，每个元组都会创建一个connection连接对象，如果有几万条数据，这样性能如何？
    // 但是又不能把创建连接代码提出来，否则，造成对象没有序列化异常
    // 所以最好的解决方案是foreachPartition
    // 以分区为单位去执行insert方法
    // 如果是2个分区，只会创建2个connection
    listRdd.foreachPartition(insert)

    sc.stop()
  }

  def insert(iterator: Iterator[(String, Int)]): Unit = {

    // 创建连接驱动实例
    Class.forName(driver).newInstance()
    // 获取连接
    val conn: Connection = DriverManager.getConnection(url, username, password)

    iterator.foreach(data => {
      val prst: PreparedStatement = conn.prepareStatement("insert into user(name, age) values (?, ?)")
      prst.setString(1, data._1)
      prst.setInt(2, data._2)

      prst.executeUpdate()

      prst.close()
    })

    conn.close()
  }
}
